package com.example.demo.entity;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import javax.annotation.PostConstruct;
import java.util.List;

import static java.util.Objects.hash;

/**
 * Redis布隆过滤器：
 * 可扩展Bloom过滤器：一单Bloom过滤器到达容量，就会再其上创建一个新的过滤器
 * 不存在重启失效或者定时任务维护的成本，可基于Google实现的布隆过滤器需要启动之后初始化布隆过滤器
 * <p>
 * 缺点：需要网络IO，性能比Google布隆过滤器低
 *
 * @author Code Farmer
 * @date 2020/4/26 15:36
 */
@ConfigurationProperties("bloom.filter")
@Component
public class RedisBloomFilter {

    //预计插入量
    private long expectedInsertions;

    //可接受的容错率
    private double fpp;

    @Autowired
    private RedisTemplate redisTemplate;

    //bit数组长度
    private long numBits;
    //hash函数数量
    private int numHashFunctions;

    public long getExpectedInsertions() {
        return expectedInsertions;
    }

    public void setExpectedInsertions(long expectedInsertions) {
        this.expectedInsertions = expectedInsertions;
    }

    public double getFpp() {
        return fpp;
    }

    public void setFpp(double fpp) {
        this.fpp = fpp;
    }

    @PostConstruct
    public void init() {
        this.numBits = optimalNumOfBits(expectedInsertions, fpp);
        this.numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);
    }

    //计算bit数组长度
    private long optimalNumOfBits(long n, double p) {
        if (p == 0.0D) {
            p = Double.MIN_VALUE;
        }

        return (long) ((double) (-n) * Math.log(p) / (Math.log(2.0D) * Math.log(2.0D)));
    }

    //计算hash函数个数
    private int optimalNumOfHashFunctions(long n, double m) {
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }

    /**
     * 判断key是否存在于集合中  是返回true  否则返回false
     */
    public boolean isExist(String key) {
        long[] indexs = getIndexs(key);
        List list = redisTemplate.executePipelined(new RedisCallback<Object>() {

            @Nullable
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                redisConnection.openPipeline();
                for (long index : indexs) {
                    redisConnection.setBit("bf:taibai".getBytes(), index, true);
                }
                return null;
            }
        });
        return !list.contains(false);
    }

    /**
     * 将key存入redis bitmap
     */
    public void put(String key) {
        long[] indexs = getIndexs(key);
        List list = redisTemplate.executePipelined(new RedisCallback<Object>() {

            @Nullable
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                redisConnection.openPipeline();
                for (long index : indexs) {
                    redisConnection.setBit("bf:taibai".getBytes(), index, true);
                }
                redisConnection.close();
                return null;
            }
        });
    }

    private long[] getIndexs(String key) {
        long hash1 = hash(key);
        long hash2 = hash1 >>> 16;
        long[] result = new long[numHashFunctions];
        for (int i = 0; i < numHashFunctions; i++) {
            long combineHash = hash1 + i * hash2;
            if (combineHash < 0) {
                combineHash = ~combineHash;
            }
            result[i] = combineHash % numBits;
        }
        return result;
    }

}
